package com.swak.reactivex.future;

import static java.util.concurrent.atomic.AtomicReferenceFieldUpdater.newUpdater;

import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executors;
import java.util.concurrent.RunnableFuture;
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;

/**
 * CompletableFutureTask.
 * 
 * @author 618lf
 */
@SuppressWarnings({ "unchecked", "rawtypes" })
public class CompletableFutureTask<V> extends CompletableFuture<V> implements RunnableFuture<V> {
	private static final AtomicReferenceFieldUpdater<CompletableFutureTask, Thread> RUNNER = newUpdater(
			CompletableFutureTask.class, Thread.class, "runner");

	private final Callable<V> callable;
	private volatile Thread runner;

	public CompletableFutureTask(Callable<V> callable) {
		super();
		this.callable = callable;
	}

	public CompletableFutureTask(Runnable runnable) {
		super();
		this.callable = Executors.callable(runnable, null);
	}

	public CompletableFutureTask(Runnable runnable, V result) {
		super();
		this.callable = Executors.callable(runnable, result);
	}

	@Override
	public void run() {
		if (isDone()) {
			return;
		}

		if (runner != null || !RUNNER.compareAndSet(this, null, Thread.currentThread())) {
			// prevents concurrent calls to run
			return;
		}

		try {
			try {
				Object result = runTask();
				complete((V) result);
			} catch (Throwable ex) {
				completeExceptionally(ex);
			}
		} finally {
			// runner must be non-null until state is settled in setResult() to
			// prevent concurrent calls to run()
			runner = null;
		}
	}

	protected V runTask() throws Exception {
		Callable c = callable;
		if (c != null) {
			return (V) c.call();
		}
		return null;
	}

	@Override
	public boolean cancel(boolean mayInterruptIfRunning) {
		boolean result = super.cancel(mayInterruptIfRunning);
		// additionally handle the interruption of the executing thread
		if (result && mayInterruptIfRunning) {
			Thread executingThread = runner;
			if (executingThread != null) {
				executingThread.interrupt();
			}
		}
		return result;
	}
}
